use std::{
    any::Any,
    sync::{Arc, Mutex},
};

use bevy_tasks::{ComputeTaskPool, Scope, TaskPool, ThreadExecutor};
use bevy_utils::default;
use bevy_utils::syncunsafecell::SyncUnsafeCell;
#[cfg(feature = "trace")]
use bevy_utils::tracing::{info_span, Instrument};
use std::panic::AssertUnwindSafe;

use async_channel::{Receiver, Sender};
use fixedbitset::FixedBitSet;

use crate::{
    archetype::ArchetypeComponentId,
    prelude::Resource,
    query::Access,
    schedule::{is_apply_deferred, BoxedCondition, ExecutorKind, SystemExecutor, SystemSchedule},
    system::BoxedSystem,
    world::{unsafe_world_cell::UnsafeWorldCell, World},
};

use crate as bevy_ecs;

/// A funky borrow split of [`SystemSchedule`] required by the [`MultiThreadedExecutor`].
struct SyncUnsafeSchedule<'a> {
    systems: &'a [SyncUnsafeCell<BoxedSystem>],
    conditions: Conditions<'a>,
}

struct Conditions<'a> {
    /// all condition for system
    /// system_conditions[system index] 系统的所有条件
    system_conditions: &'a mut [Vec<BoxedCondition>],
    /// set_conditions[set index] system set 对应的所有条件, 一次运行只会评估一次
    set_conditions: &'a mut [Vec<BoxedCondition>],
    /// system 依赖的所有 set_conditions 索引
    /// [set_idx] = sets_with_conditions_of_systems[system_index]
    sets_with_conditions_of_systems: &'a [FixedBitSet],
    /// systems_in_sets_with_conditions[set_idx] = 依赖此集合条件的所有system index
    systems_in_sets_with_conditions: &'a [FixedBitSet],
}

impl SyncUnsafeSchedule<'_> {
    fn new(schedule: &mut SystemSchedule) -> SyncUnsafeSchedule<'_> {
        SyncUnsafeSchedule {
            systems: SyncUnsafeCell::from_mut(schedule.systems.as_mut_slice()).as_slice_of_cells(),
            conditions: Conditions {
                system_conditions: &mut schedule.system_conditions,
                set_conditions: &mut schedule.set_conditions,
                sets_with_conditions_of_systems: &schedule.sets_with_conditions_of_systems,
                systems_in_sets_with_conditions: &schedule.systems_in_sets_with_conditions,
            },
        }
    }
}

/// Per-system data used by the [`MultiThreadedExecutor`].
/// Copied here because it can't be read from the system when it's running.
struct SystemTaskMetadata {
    /// The [`ArchetypeComponentId`] access of the system.
    /// 使用到的 archetype componet id
    archetype_component_access: Access<ArchetypeComponentId>,
    /// Indices of the systems that directly depend on the system.
    /// 哪些 system index 直接依赖本system
    dependents: Vec<usize>,
    /// Is `true` if the system does not access `!Send` data.
    is_send: bool,
    /// Is `true` if the system is exclusive.
    is_exclusive: bool,
}

/// The result of running a system that is sent across a channel.
struct SystemResult {
    /// 已完成的system index
    system_index: usize,
    success: bool,
}

/// Runs the schedule using a thread pool. Non-conflicting systems can run in parallel.
pub struct MultiThreadedExecutor {
    /// Sends system completion events.
    sender: Sender<SystemResult>,
    /// Receives system completion events.
    receiver: Receiver<SystemResult>,
    /// Metadata for scheduling and running system tasks.
    system_task_metadata: Vec<SystemTaskMetadata>,
    /// Union of the accesses of all currently running systems.
    active_access: Access<ArchetypeComponentId>,
    /// Returns `true` if a system with non-`Send` access is running.
    local_thread_running: bool,
    /// Returns `true` if an exclusive system is running.
    exclusive_running: bool,
    /// The number of systems expected to run.
    num_systems: usize,
    /// The number of systems that are running.
    num_running_systems: usize,
    /// The number of systems that have completed.
    num_completed_systems: usize,
    /// The number of dependencies each system has that have not completed.
    num_dependencies_remaining: Vec<usize>,
    /// System sets whose conditions have been evaluated.
    evaluated_sets: FixedBitSet,
    /// Systems that have no remaining dependencies and are waiting to run.
    ready_systems: FixedBitSet,
    /// copy of `ready_systems`
    ready_systems_copy: FixedBitSet,
    /// Systems that are running.
    running_systems: FixedBitSet,
    /// Systems that got skipped.
    skipped_systems: FixedBitSet,
    /// Systems whose conditions have been evaluated and were run or skipped.
    completed_systems: FixedBitSet,
    /// Systems that have run but have not had their buffers applied.
    unapplied_systems: FixedBitSet,
    /// Setting when true applies deferred system buffers after all systems have run
    apply_final_deferred: bool,
    /// When set, tells the executor that a thread has panicked.
    panic_payload: Arc<Mutex<Option<Box<dyn Any + Send>>>>,
    /// When set, stops the executor from running any more systems.
    stop_spawning: bool,
}

impl Default for MultiThreadedExecutor {
    fn default() -> Self {
        Self::new()
    }
}

impl SystemExecutor for MultiThreadedExecutor {
    fn kind(&self) -> ExecutorKind {
        ExecutorKind::MultiThreaded
    }

    fn set_apply_final_deferred(&mut self, value: bool) {
        self.apply_final_deferred = value;
    }

    fn init(&mut self, schedule: &SystemSchedule) {
        // pre-allocate space
        let sys_count = schedule.system_ids.len();
        let set_count = schedule.set_ids.len();

        // 按system个数创建channel
        let (tx, rx) = async_channel::bounded(sys_count.max(1));

        self.sender = tx;
        self.receiver = rx;
        self.evaluated_sets = FixedBitSet::with_capacity(set_count);
        self.ready_systems = FixedBitSet::with_capacity(sys_count);
        self.ready_systems_copy = FixedBitSet::with_capacity(sys_count);
        self.running_systems = FixedBitSet::with_capacity(sys_count);
        self.completed_systems = FixedBitSet::with_capacity(sys_count);
        self.skipped_systems = FixedBitSet::with_capacity(sys_count);
        self.unapplied_systems = FixedBitSet::with_capacity(sys_count);

        self.system_task_metadata = Vec::with_capacity(sys_count);
        for index in 0..sys_count {
            self.system_task_metadata.push(SystemTaskMetadata {
                archetype_component_access: default(),
                dependents: schedule.system_dependents[index].clone(),
                is_send: schedule.systems[index].is_send(),
                is_exclusive: schedule.systems[index].is_exclusive(),
            });
        }

        self.num_dependencies_remaining = Vec::with_capacity(sys_count);
    }

    // 运行一次
    fn run(&mut self, schedule: &mut SystemSchedule, world: &mut World) {
        // reset counts
        self.num_systems = schedule.systems.len();
        if self.num_systems == 0 {
            return;
        }
        self.num_running_systems = 0;
        self.num_completed_systems = 0;
        self.num_dependencies_remaining.clear();
        self.num_dependencies_remaining
            .extend_from_slice(&schedule.system_dependencies);

        for (system_index, dependencies) in self.num_dependencies_remaining.iter_mut().enumerate() {
            // 没有任何依赖的系统，放到就绪系统集
            if *dependencies == 0 {
                self.ready_systems.insert(system_index);
            }
        }

        // 主线程执行器
        let thread_executor = world
            .get_resource::<MainThreadExecutor>()
            .map(|e| e.0.clone());
        let thread_executor = thread_executor.as_deref();

        let SyncUnsafeSchedule {
            systems,
            mut conditions,
        } = SyncUnsafeSchedule::new(schedule);

        // 阻塞，进行system 调度
        ComputeTaskPool::init(TaskPool::default).scope_with_executor(
            false,
            thread_executor,
            |scope| {
                // the executor itself is a `Send` future so that it can run
                // alongside systems that claim the local thread
                let executor = async {
                    let world_cell = world.as_unsafe_world_cell();
                    // 持续将所有system都调度完
                    while self.num_completed_systems < self.num_systems {
                        // SAFETY:
                        // - self.ready_systems does not contain running systems.
                        // - `world_cell` has mutable access to the entire world.
                        unsafe {
                            self.spawn_system_tasks(scope, systems, &mut conditions, world_cell);
                        }

                        // 有系统正在运行
                        if self.num_running_systems > 0 {
                            // wait for systems to complete
                            // 至少等待一个system完成
                            if let Ok(result) = self.receiver.recv().await {
                                self.finish_system_and_handle_dependents(result);
                            } else {
                                panic!("Channel closed unexpectedly!");
                            }

                            // 上面保证至少一个system完成了
                            // 这里检测是否还有更多的system完成了
                            while let Ok(result) = self.receiver.try_recv() {
                                self.finish_system_and_handle_dependents(result);
                            }

                            self.rebuild_active_access();
                        }
                    }
                };

                #[cfg(feature = "trace")]
                let executor_span = info_span!("multithreaded executor");
                #[cfg(feature = "trace")]
                let executor = executor.instrument(executor_span);
                // 启动任务调度任务
                scope.spawn(executor);
            },
        );

        if self.apply_final_deferred {
            // Do one final apply buffers after all systems have completed
            // Commands should be applied while on the scope's thread, not the executor's thread
            let res = apply_deferred(&self.unapplied_systems, systems, world);
            if let Err(payload) = res {
                let mut panic_payload = self.panic_payload.lock().unwrap();
                *panic_payload = Some(payload);
            }
            self.unapplied_systems.clear();
            debug_assert!(self.unapplied_systems.is_clear());
        }

        // check to see if there was a panic
        let mut payload = self.panic_payload.lock().unwrap();
        if let Some(payload) = payload.take() {
            std::panic::resume_unwind(payload);
        }

        debug_assert!(self.ready_systems.is_clear());
        debug_assert!(self.running_systems.is_clear());
        self.active_access.clear();
        self.evaluated_sets.clear();
        self.skipped_systems.clear();
        self.completed_systems.clear();
    }
}

impl MultiThreadedExecutor {
    /// Creates a new multi-threaded executor for use with a [`Schedule`].
    ///
    /// [`Schedule`]: crate::schedule::Schedule
    pub fn new() -> Self {
        let (sender, receiver) = async_channel::unbounded();
        Self {
            sender,
            receiver,
            system_task_metadata: Vec::new(),
            num_systems: 0,
            num_running_systems: 0,
            num_completed_systems: 0,
            num_dependencies_remaining: Vec::new(),
            active_access: default(),
            local_thread_running: false,
            exclusive_running: false,
            evaluated_sets: FixedBitSet::new(),
            ready_systems: FixedBitSet::new(),
            ready_systems_copy: FixedBitSet::new(),
            running_systems: FixedBitSet::new(),
            skipped_systems: FixedBitSet::new(),
            completed_systems: FixedBitSet::new(),
            unapplied_systems: FixedBitSet::new(),
            apply_final_deferred: true,
            panic_payload: Arc::new(Mutex::new(None)),
            stop_spawning: false,
        }
    }

    /// # Safety
    /// - Caller must ensure that `self.ready_systems` does not contain any systems that
    ///   have been mutably borrowed (such as the systems currently running).
    /// - `world_cell` must have permission to access all world data (not counting
    ///   any world data that is claimed by systems currently running on this executor).
    /// 调度system运行,可以一次并行运行多个system
    unsafe fn spawn_system_tasks<'scope>(
        &mut self,
        scope: &Scope<'_, 'scope, ()>,
        systems: &'scope [SyncUnsafeCell<BoxedSystem>],
        conditions: &mut Conditions,
        world_cell: UnsafeWorldCell<'scope>,
    ) {
        // 独占系统正在运行，不能运行其它system
        if self.exclusive_running {
            return;
        }

        // can't borrow since loop mutably borrows `self`
        // 下面地方for 里的 skip_system_and_signal_dependents 会增加 ready_systems，所以这里迭代前复制下
        let mut ready_systems = std::mem::take(&mut self.ready_systems_copy);
        ready_systems.clear();
        ready_systems.union_with(&self.ready_systems);

        // 遍历准备就绪的系统
        for system_index in ready_systems.ones() {
            // 就绪的系统一定没有正在运行
            assert!(!self.running_systems.contains(system_index));
            // SAFETY: Caller assured that these systems are not running.
            // Therefore, no other reference to this system exists and there is no aliasing.
            let system = unsafe { &mut *systems[system_index].get() };

            // 此系统不能运行
            if !self.can_run(system_index, system, conditions, world_cell) {
                // NOTE: exclusive systems with ambiguities are susceptible to
                // being significantly displaced here (compared to single-threaded order)
                // if systems after them in topological order can run
                // if that becomes an issue, `break;` if exclusive system
                // 继续检测下个system 
                continue;
            }

            // 后续可能运行，先从准备就绪里移除
            self.ready_systems.set(system_index, false);

            // SAFETY: `can_run` returned true, which means that:
            // - It must have called `update_archetype_component_access` for each run condition.
            // - There can be no systems running whose accesses would conflict with any conditions.
            // 不应该运行
            if !self.should_run(system_index, system, conditions, world_cell) {
                // 跳过此系统的运行，通知所有依赖系统
                self.skip_system_and_signal_dependents(system_index);
                // 继续检测下个system 
                continue;
            }

            // 记录到正在运行系统集中
            self.running_systems.insert(system_index);
            self.num_running_systems += 1;

            // 此系统是独占系统
            if self.system_task_metadata[system_index].is_exclusive {
                // SAFETY: `can_run` returned true for this system, which means
                // that no other systems currently have access to the world.
                let world = unsafe { world_cell.world_mut() };
                // SAFETY: `can_run` returned true for this system,
                // which means no systems are currently borrowed.
                unsafe {
                    self.spawn_exclusive_system_task(scope, system_index, systems, world);
                }
                // 运行的是独占系统，没必要再检测其它system，因为其它system必然不能运行
                break;
            }

            // 普通系统，从任务池里运行
            // SAFETY:
            // - No other reference to this system exists.
            // - `can_run` has been called, which calls `update_archetype_component_access` with this system.
            // - `can_run` returned true, so no systems with conflicting world access are running.
            unsafe {
                self.spawn_system_task(scope, system_index, systems, world_cell);
            }
        }

        // give back
        self.ready_systems_copy = ready_systems;
    }

    /// 检测 system[system_index] 对应的 system 是否可以运行，
    /// 检测system 及所有依赖的condition 是否和运行中的system冲突
    fn can_run(
        &mut self,
        system_index: usize,
        system: &mut BoxedSystem,
        conditions: &mut Conditions,
        world: UnsafeWorldCell,
    ) -> bool {
        let system_meta = &self.system_task_metadata[system_index];
        // 本system是独占的，当前有system正在运行，本独占system不能运行
        if system_meta.is_exclusive && self.num_running_systems > 0 {
            return false;
        }

        // 本system是需要在主线程运行的，但是主线程上已有system在运行中了
        if !system_meta.is_send && self.local_thread_running {
            return false;
        }

        // TODO: an earlier out if world's archetypes did not change
        // 遍历本system依赖的尚未评估的 system set 条件
        for set_idx in conditions.sets_with_conditions_of_systems[system_index]
            .difference(&self.evaluated_sets)
        {
            // all condition of system set 
            // 遍历此system set 的所有 condition system
            for condition in &mut conditions.set_conditions[set_idx] {
                condition.update_archetype_component_access(world);
                // condition of set is not compatible with running system, can't run
                // 与正在运行中的system有冲突
                if !condition
                    .archetype_component_access()
                    .is_compatible(&self.active_access)
                {
                    return false;
                }
            }
        }

        // 遍历此system的所有condition system
        for condition in &mut conditions.system_conditions[system_index] {
            condition.update_archetype_component_access(world);
            // system condition not compatible with running system,can't run
            // 与正在运行中的system有冲突
            if !condition
                .archetype_component_access()
                .is_compatible(&self.active_access)
            {
                return false;
            }
        }

        // 此系统未被跳过
        if !self.skipped_systems.contains(system_index) {
            system.update_archetype_component_access(world);
            // access not compatible with running system , can't run in same time
            // 与正在运行中的system有冲突
            if !system
                .archetype_component_access()
                .is_compatible(&self.active_access)
            {
                return false;
            }

            // 同步system.archetype_component_access到system_task_metadata[system_index].archetype_component_access
            // PERF: use an optimized clear() + extend() operation
            let meta_access =
                &mut self.system_task_metadata[system_index].archetype_component_access;
            meta_access.clear();
            // reset system_task_metadata[system_index].archetype_component_access
            meta_access.extend(system.archetype_component_access());
        }
        // todo: ?此系统被跳过,可以运行

        true
    }

    /// # Safety
    /// * `world` must have permission to read any world data required by
    ///   the system's conditions: this includes conditions for the system
    ///   itself, and conditions for any of the system's sets.
    /// * `update_archetype_component` must have been called with `world`
    ///   for each run condition in `conditions`.
    /// 返回system是否满足配置的运行条件
    unsafe fn should_run(
        &mut self,
        system_index: usize,
        _system: &BoxedSystem,
        conditions: &mut Conditions,
        world: UnsafeWorldCell,
    ) -> bool {
        // 是否可以运行
        // 跳过的不需要运行
        let mut should_run = !self.skipped_systems.contains(system_index);

        // 遍历system依赖的所有未评估过的 system set's condition
        // 即使跳过的system，也评估其所有系统集条件
        for set_idx in conditions.sets_with_conditions_of_systems[system_index].ones() {
            // 依赖的 system set 已经评估过了
            if self.evaluated_sets.contains(set_idx) {
                continue;
            }

            // Evaluate the system set's conditions.
            // SAFETY:
            // - The caller ensures that `world` has permission to read any data
            //   required by the conditions.
            // - `update_archetype_component_access` has been called for each run condition.
            let set_conditions_met =
                evaluate_and_fold_conditions(&mut conditions.set_conditions[set_idx], world);

            // system set condition 不满足,不能运行
            if !set_conditions_met {
                // 把依赖此集合条件的system都跳过
                self.skipped_systems
                    .union_with(&conditions.systems_in_sets_with_conditions[set_idx]);
            }

            should_run &= set_conditions_met;
            // 记录system set 已经被评估过了，同一个不会重复评估,
            // 如果集合条件不满足，所有依赖此集合的system都被加入到skipped_systems了，后续评估结果不会导致可以运行
            self.evaluated_sets.insert(set_idx);
        }

        // Evaluate the system's conditions.
        // SAFETY:
        // - The caller ensures that `world` has permission to read any data
        //   required by the conditions.
        // - `update_archetype_component_access` has been called for each run condition.
        let system_conditions_met =
            evaluate_and_fold_conditions(&mut conditions.system_conditions[system_index], world);

        // 系统条件不满足
        if !system_conditions_met {
            // 记录此系统被跳过
            self.skipped_systems.insert(system_index);
        }

        // 系统条件不满足的系统，不能运行
        should_run &= system_conditions_met;

        should_run
    }

    /// # Safety
    /// - Caller must not alias systems that are running.
    /// - `world` must have permission to access the world data
    ///   used by the specified system.
    /// - `update_archetype_component_access` must have been called with `world`
    ///   on the system assocaited with `system_index`.
    /// 在scope任务池中异步启动system
    unsafe fn spawn_system_task<'scope>(
        &mut self,
        scope: &Scope<'_, 'scope, ()>,
        system_index: usize,
        systems: &'scope [SyncUnsafeCell<BoxedSystem>],
        world: UnsafeWorldCell<'scope>,
    ) {
        // SAFETY: this system is not running, no other reference exists
        let system = unsafe { &mut *systems[system_index].get() };

        #[cfg(feature = "trace")]
        let task_span = info_span!("system_task", name = &*system.name());
        #[cfg(feature = "trace")]
        let system_span = info_span!("system", name = &*system.name());

        let sender = self.sender.clone();
        let panic_payload = self.panic_payload.clone();
        // 构造异步任务
        let task = async move {
            #[cfg(feature = "trace")]
            let system_guard = system_span.enter();
            // 防止system panic 影响整个进程
            let res = std::panic::catch_unwind(AssertUnwindSafe(|| {
                // SAFETY:
                // - The caller ensures that we have permission to
                // access the world data used by the system.
                // - `update_archetype_component_access` has been called.
                // 运行此system
                unsafe { system.run_unsafe((), world) };
            }));
            #[cfg(feature = "trace")]
            drop(system_guard);
            // tell the executor that the system finished
            // system run finished, notify executor
            sender
                .try_send(SystemResult {
                    system_index,
                    success: res.is_ok(),
                })
                .unwrap_or_else(|error| unreachable!("{}", error));
            if let Err(payload) = res {
                eprintln!("Encountered a panic in system `{}`!", &*system.name());
                // set the payload to propagate the error
                {
                    let mut panic_payload = panic_payload.lock().unwrap();
                    *panic_payload = Some(payload);
                }
            }
        };

        #[cfg(feature = "trace")]
        let task = task.instrument(task_span);

        let system_meta = &self.system_task_metadata[system_index];
        // 此system将要运行，将archetype_component_access合并到运行中active_access
        self.active_access
            .extend(&system_meta.archetype_component_access);

        if system_meta.is_send {
            // 在池里异步启动任务
            scope.spawn(task);
        } else {
            // 在主线程启动任务
            self.local_thread_running = true;
            scope.spawn_on_external(task);
        }
    }

    /// # Safety
    /// Caller must ensure no systems are currently borrowed.
    /// 启动独占系统 systems[system_index] 任务
    unsafe fn spawn_exclusive_system_task<'scope>(
        &mut self,
        scope: &Scope<'_, 'scope, ()>,
        system_index: usize,
        systems: &'scope [SyncUnsafeCell<BoxedSystem>],
        world: &'scope mut World,
    ) {
        // SAFETY: this system is not running, no other reference exists
        let system = unsafe { &mut *systems[system_index].get() };

        #[cfg(feature = "trace")]
        let task_span = info_span!("system_task", name = &*system.name());
        #[cfg(feature = "trace")]
        let system_span = info_span!("system", name = &*system.name());

        // 完成后，通过此channal通知
        let sender = self.sender.clone();
        let panic_payload = self.panic_payload.clone();
        if is_apply_deferred(system) {
            // TODO: avoid allocation
            // 复制 self.unapplied_systems
            let unapplied_systems = self.unapplied_systems.clone();
            // 清空 self.unapplied_systems
            self.unapplied_systems.clear();
            // 创建buffer应用任务
            let task = async move {
                #[cfg(feature = "trace")]
                let system_guard = system_span.enter();
                // 应用buffer
                let res = apply_deferred(&unapplied_systems, systems, world);
                #[cfg(feature = "trace")]
                drop(system_guard);
                // tell the executor that the system finished
                sender
                    .try_send(SystemResult {
                        system_index,
                        success: res.is_ok(),
                    })
                    .unwrap_or_else(|error| unreachable!("{}", error));
                if let Err(payload) = res {
                    // set the payload to propagate the error
                    let mut panic_payload = panic_payload.lock().unwrap();
                    *panic_payload = Some(payload);
                }
            };

            #[cfg(feature = "trace")]
            let task = task.instrument(task_span);
            // 在池中启动任务
            scope.spawn_on_scope(task);
        } else {
            // 普通system
            let task = async move {
                #[cfg(feature = "trace")]
                let system_guard = system_span.enter();
                let res = std::panic::catch_unwind(AssertUnwindSafe(|| {
                    system.run((), world);
                }));
                #[cfg(feature = "trace")]
                drop(system_guard);
                // tell the executor that the system finished
                sender
                    .try_send(SystemResult {
                        system_index,
                        success: res.is_ok(),
                    })
                    .unwrap_or_else(|error| unreachable!("{}", error));
                if let Err(payload) = res {
                    eprintln!(
                        "Encountered a panic in exclusive system `{}`!",
                        &*system.name()
                    );
                    // set the payload to propagate the error
                    let mut panic_payload = panic_payload.lock().unwrap();
                    *panic_payload = Some(payload);
                }
            };

            #[cfg(feature = "trace")]
            let task = task.instrument(task_span);
            // 在池中启动任务
            scope.spawn_on_scope(task);
        }

        // 独占系统运行中
        self.exclusive_running = true;
        // 独占系统在主线程运行
        self.local_thread_running = true;
    }

    /// 一个system run完成了,更新数据
    fn finish_system_and_handle_dependents(&mut self, result: SystemResult) {
        let SystemResult {
            system_index,
            success,
        } = result;

        // 独占系统运行完成
        if self.system_task_metadata[system_index].is_exclusive {
            // 当前没有在运行的独占系统
            self.exclusive_running = false;
        }

        // 主线程system运行完成
        if !self.system_task_metadata[system_index].is_send {
            // 没有运行的主线程system
            self.local_thread_running = false;
        }

        debug_assert!(self.num_running_systems >= 1);
        self.num_running_systems -= 1;
        self.num_completed_systems += 1;
        // 从运行中system里移除
        self.running_systems.set(system_index, false);
        // 记录到已完成
        self.completed_systems.insert(system_index);
        // 刚完成的system，尚未apply buffer
        self.unapplied_systems.insert(system_index);

        self.signal_dependents(system_index);

        // system 运行失败了?
        if !success {
            self.stop_spawning_systems();
        }
    }

    /// system_index 的 system 不应该运行,通知所有依赖此system的system,
    /// 记录未已完成，后续其它依赖此system的system就可以启动了
    fn skip_system_and_signal_dependents(&mut self, system_index: usize) {
        self.num_completed_systems += 1;
        self.completed_systems.insert(system_index);
        self.signal_dependents(system_index);
    }

    /// 把所有直接依赖 system_index system 的依赖计数-1，
    /// 将依赖都完成的放到待运行集合里 ready_systems 
    fn signal_dependents(&mut self, system_index: usize) {
        // 遍历所有直接依赖 system_index 的 system dep_idx
        for &dep_idx in &self.system_task_metadata[system_index].dependents {
            // 获取 dep_idx system 剩余未完成的的直接依赖计数
            let remaining = &mut self.num_dependencies_remaining[dep_idx];
            debug_assert!(*remaining >= 1);
            // 依赖system_index 已经完成，减少直接依赖计数
            *remaining -= 1;
            // dep_idx 不再依赖其它system，并且是未执行完成的系统
            if *remaining == 0 && !self.completed_systems.contains(dep_idx) {
                // 将 dep_idx 放到等待运行的系统集合里
                self.ready_systems.insert(dep_idx);
            }
        }
    }

    fn stop_spawning_systems(&mut self) {
        if !self.stop_spawning {
            self.num_systems = self.num_completed_systems + self.num_running_systems;
            self.stop_spawning = true;
        }
    }

    // 用正在运行的系统 构建active_access
    fn rebuild_active_access(&mut self) {
        self.active_access.clear();
        for index in self.running_systems.ones() {
            let system_meta = &self.system_task_metadata[index];
            self.active_access
                .extend(&system_meta.archetype_component_access);
        }
    }
}

/// 应用延迟指令
/// unapplied_systems 里置1的system index，执行 system.apply_deferred(world)
fn apply_deferred(
    unapplied_systems: &FixedBitSet,
    systems: &[SyncUnsafeCell<BoxedSystem>],
    world: &mut World,
) -> Result<(), Box<dyn std::any::Any + Send>> {
    for system_index in unapplied_systems.ones() {
        // SAFETY: none of these systems are running, no other references exist
        let system = unsafe { &mut *systems[system_index].get() };
        let res = std::panic::catch_unwind(AssertUnwindSafe(|| {
            system.apply_deferred(world);
        }));
        if let Err(payload) = res {
            eprintln!(
                "Encountered a panic when applying buffers for system `{}`!",
                &*system.name()
            );
            return Err(payload);
        }
    }
    Ok(())
}

/// # Safety
/// - `world` must have permission to read any world data
///   required by `conditions`.
/// - `update_archetype_component_access` must have been called
///   with `world` for each condition in `conditions`.
unsafe fn evaluate_and_fold_conditions(
    conditions: &mut [BoxedCondition],
    world: UnsafeWorldCell,
) -> bool {
    // not short-circuiting is intentional
    #[allow(clippy::unnecessary_fold)]
    conditions
        .iter_mut()
        .map(|condition| {
            #[cfg(feature = "trace")]
            let _condition_span = info_span!("condition", name = &*condition.name()).entered();
            // SAFETY: The caller ensures that `world` has permission to
            // access any data required by the condition.
            unsafe { condition.run_unsafe((), world) }
        })
        .fold(true, |acc, res| acc && res)
}

/// New-typed [`ThreadExecutor`] [`Resource`] that is used to run systems on the main thread
#[derive(Resource, Clone)]
pub struct MainThreadExecutor(pub Arc<ThreadExecutor<'static>>);

impl Default for MainThreadExecutor {
    fn default() -> Self {
        Self::new()
    }
}

impl MainThreadExecutor {
    /// Creates a new executor that can be used to run systems on the main thread.
    pub fn new() -> Self {
        MainThreadExecutor(TaskPool::get_thread_executor())
    }
}
